package cn.itcast.cep;

import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Arrays;
import java.util.List;
import java.util.Map;

//连续模式：匹配到的数据需要是连续出现的，如果不连续就会中断匹配。
//需求：从数据源中依次提取"c","a","b"元素
public class ConsecutiveDemo {
    public static void main(String[] args) throws Exception {
        //todo 1.获取流处理执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //todo 2.设置但并行度
        env.setParallelism(1);
        //todo 3.加载数据源
        DataStreamSource<String> source = env.fromElements("c", "d", "a", "a", "a", "d", "a", "b");
        //todo 4.设置匹配模式，匹配"c","a","b"
        //todo   多次匹配"a"：组合模式
        Pattern<String, String> pattern = Pattern.<String>begin("begin")
                .where(new SimpleCondition<String>() {
                    @Override
                    public boolean filter(String s) throws Exception {
                        return s.equals("c");
                    }
                })
                .followedBy("middle")
                .where(new SimpleCondition<String>() {
                    @Override
                    public boolean filter(String s) throws Exception {
                        return s.equals("a");
                    }
                })
                .oneOrMore() //默认一次
                .consecutive() //连续模式：匹配到的数据需要是连续出现的，如果不连续就会中断匹配。
                //.allowCombinations()//允许组合:匹配到的数据可以是不连续出现的。
                .followedBy("end")
                .where(new SimpleCondition<String>() {
                    @Override
                    public boolean filter(String s) throws Exception {
                        return s.equals("b");
                    }
                });
        //todo 5.匹配数据提取Tuple3
        PatternStream<String> cep = CEP.pattern(source, pattern);
        cep.select(new PatternSelectFunction<String, Object>() {
            @Override
            public Object select(Map<String, List<String>> map) throws Exception {
                List<String> begin = map.get("begin");
                List<String> middle = map.get("middle");
                List<String> end = map.get("end");
                return Tuple3.of(begin,middle,end);
            }
        }).print();
        //todo 6.数据打印
        //todo 7.触发执行
        env.execute();
    }
}
